package blue.mqtt4j.internal.client.impl;

import blue.mqtt4j.client.ConnectionListener;
import blue.mqtt4j.client.ConsumerListener;
import blue.mqtt4j.client.MqttClient;
import blue.mqtt4j.client.MqttMessage;
import blue.mqtt4j.client.MqttOptions;
import blue.mqtt4j.client.MqttTopic;
import blue.mqtt4j.client.Producer;
import blue.mqtt4j.client.ProducerListener;
import blue.mqtt4j.client.codec.Decoder;
import blue.mqtt4j.client.codec.Encoder;
import blue.mqtt4j.core.topic.MqttTopicUtil;
import blue.mqtt4j.core.util.AssertUtil;
import blue.mqtt4j.core.util.WaitUtil;
import blue.mqtt4j.internal.client.net.NettyClient;
import blue.mqtt4j.internal.client.net.NettyProducerListener;
import blue.mqtt4j.internal.client.task.TaskExecutor;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author Jin Zheng
 * @since 1.0 2020-09-30
 */
public class DefaultMqttClient implements MqttClient
{
	private static Logger logger = LoggerFactory.getLogger(DefaultMqttClient.class);

	private final String broker;
	private final MqttOptions options;
	private final DefaultProducer producer;
	private final NettyClient nettyClient;
	private final TaskExecutor executor;
	private final ReceiveHandler container;
	private ConnectionListener listener;
	private CountDownLatch latch;

	public DefaultMqttClient(String broker, MqttOptions options)
	{
		this.broker = broker;
		this.options = options;
		this.producer = new DefaultProducer(this);
		this.nettyClient = new NettyClient(broker, options, this);
		this.executor = new TaskExecutor(options.maxThreads());
		this.container = new ReceiveHandler(executor, nettyClient);
	}

	@Override
	public void connect()
	{
		this.latch = new CountDownLatch(1);
		nettyClient.connect();
		WaitUtil.await(this.latch);
		this.latch = null;
	}

	@Override
	public void close()
	{
		this.latch = new CountDownLatch(1);
		nettyClient.disconnect();
		WaitUtil.await(this.latch, options.timeoutMillis(), TimeUnit.MILLISECONDS);
		this.latch = null;
		this.executor.shutdown();
	}

	@Override
	public Producer getProducer()
	{
		return producer;
	}

	@Override
	public void subscribe(List<MqttTopic> topicList, Decoder decoder, ConsumerListener listener)
	{
		AssertUtil.isEmpty(topicList, "MqttTopic");
		AssertUtil.isNull(listener, "ConsumerListener");
		for (MqttTopic topic : topicList)
		{
			MqttTopicUtil.validate(topic.topic(), true);
		}

		nettyClient.subscribe(topicList);
		container.add(topicList, decoder, listener);
	}

	@Override
	public void unsubscribe(String topic)
	{
		MqttTopicUtil.validate(topic, true);

		int messageId = nettyClient.unsubscribe(topic);
		container.addUnsubscribeTopic(messageId, new Topic(topic));
	}

	public void unsubAck(int messageId)
	{
		container.remove(messageId);
	}

	public void receive(MqttPublishMessage message)
	{
		container.receive(message);
	}

	public void pubRel(int messageId)
	{
		container.pubRel(messageId);
	}

	public void publish(MqttMessage message, Encoder encoder, ProducerListener listener, boolean async)
	{
		byte[] bytes = encoder.encode(message.payload());
		NettyProducerListener futureListener = listener == null ? null :
				new NettyProducerListener(executor, message, listener, async);
		nettyClient.publish(message.topic(), message.qos(), bytes, futureListener);
	}

	public void onConnected(boolean sessionPresent)
	{
		logger.info("MQTT connected: {}, sessionPresent: {}", broker, sessionPresent);
		if (!sessionPresent)
		{
			Set<MqttTopic> topicSet = container.topicSet();
			nettyClient.subscribe(topicSet);
		}
		this.producer.setConnectState(true);
		if (listener != null)
		{
			executor.execute(() -> listener.onConnected(broker, this));
		}
		if (this.latch != null)
		{
			this.latch.countDown();
		}
	}

	public void onDisconnected()
	{
		logger.info("MQTT disconnected: {}", broker);
		this.producer.setConnectState(false);
		if (listener != null)
		{
			executor.execute(() -> listener.onDisconnected(broker, options));
		}
		if (this.latch != null)
		{
			this.latch.countDown();
		}
	}

	@Override
	public void setConnectionListener(ConnectionListener listener)
	{
		this.listener = listener;
	}

	@Override
	public MqttOptions getOptions()
	{
		return options;
	}
}
